



# --------------------------可迭代对像--迭代器---生成器-------------------------------------

# 他们是包含关系：  可迭代对像 > 迭代器 >  生成器

# 特性	      可迭代对象	    迭代器	        生成器
# 实现__iter__	 是	            是	            是
# 实现__next__	 否	            是	            是
# 内存效率	   一般	            高	            非常高
# 状态保存	    无	            保存当前位置	保存整个执行状态
# 创建方式	   容器类型	        iter()或实现	函数+yield或生成器表达式





# 1. 可迭代对象示例 - 自定义可迭代类
class MyIterable:
    def __init__(self, start, end):
        self.start = start
        self.end = end
    
    def __iter__(self):   #----------------------------------------------------------__iter__ 返回的是一个迭代器对象-------
        return MyIterator(self.start, self.end)

# 2. 迭代器示例 - 自定义迭代器类
class MyIterator:
    def __init__(self, start, end):
        self.current = start
        self.end = end
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.current >= self.end:
            raise StopIteration
        result = self.current
        self.current += 1
        return result

# 3. 生成器示例 - 使用yield的函数
def my_generator(start, end):
    current = start
    while current < end:
        yield current
        current += 1

# 4. 生成器表达式示例
gen_expr = (x for x in range(5, 10))

# 使用示例
print("=== 可迭代对象测试 ===")
iterable = MyIterable(1, 5)
for num in iterable:  # 隐式调用iter()和next()
    print(num, end=' ')  # 输出: 1 2 3 4
print("\n")

print("=== 迭代器测试 ===")
iterator = iter(iterable)  # 显式获取迭代器
print(next(iterator))  # 输出: 1
print(next(iterator))  # 输出: 2
print(next(iterator))  # 输出: 3
print(next(iterator))  # 输出: 4
# print(next(iterator))  # 抛出StopIteration异常
print()

print("=== 生成器函数测试 ===")
gen = my_generator(10, 15)
for num in gen:
    print(num, end=' ')  # 输出: 10 11 12 13 14
print("\n")

print("=== 生成器表达式测试 ===")
for num in gen_expr:
    print(num, end=' ')  # 输出: 5 6 7 8 9
print("\n")

# 5. 内存效率对比
import sys

print("=== 内存效率对比 ===")
big_range = 1000000

# 列表推导式 - 一次性生成所有元素
list_comp = [x for x in range(big_range)]
print(f"列表推导式内存占用: {sys.getsizeof(list_comp)} bytes")

# 生成器表达式 - 按需生成元素
gen_comp = (x for x in range(big_range))
print(f"生成器表达式内存占用: {sys.getsizeof(gen_comp)} bytes")



# ---------------------- 生成器 ---------------------------------
# 生成器是创建迭代器的便捷方式，特别适合处理大数据集或无限序列

# close()
# send()
# throw()

# 遍历 range(5, 10) 产生的序列（5,6,7,8,9）
# 对每个元素 x，直接返回 x 本身
# 返回一个生成器对象 gen_expr


# 惰性求值：只有在需要时才会计算下一个值
# 内存高效：不会一次性生成所有元素
# 一次性使用：生成器通常只能迭代一次
# 不支持索引：不能像列表那样通过索引访问元素

# yield from 是 Python 3.3 引入的语法，用于简化生成器的嵌套使用

# ----------------------生成器使用方式---------------------------------
# 1. 直接迭代
for num in gen_expr:
    print(num)  # 依次输出5,6,7,8,9

# 2. 使用next()函数逐个获取
gen = (x for x in range(5, 10))
print(next(gen))  # 5
print(next(gen))  # 6
print(next(gen))  # 7

# 3. 转换为其他数据结构（会立即计算所有值）
list_from_gen = list(gen_expr)  # [5,6,7,8,9]


# 使用场景
    # 处理大文件： 计算大文件中所有行的长度，但不一次性加载所有内容  ---  超过内存的大文件 或 大的数据库查询结果
    # line_lengths = (len(line) for line in open('large_file.txt'))

    # 复杂的计算 计算平方值
    # squares = (x**2 for x in range(10))

    # 链式操作： 多个生成器表达式组合
    # nums = (x for x in range(100))
    # filtered = (x for x in nums if x % 3 == 0)
    # result = (x**2 for x in filtered)

# 生成器表达式是Python中函数式编程风格的重要工具，特别适合处理大数据流和管道操作，能显著提高程序的内存效率。

# ----------------对比列表推导式和生成器表达式---------------------------
#              生成器               列表
# 适用场景	大数据集/只需迭代一次	 需要多次访问/修改
# 内存使用	低（惰性计算）	        高（立即计算所有值）
# 返回值类型	生成器对象	          列表对象


# ------------yield-------yield from-----------------------

# 编译器将含 yield 的方法转换为一个​​状态机类​​，实现 IEnumerable 和 IEnumerator 接口。关键组件包括：
# ​​状态字段（_state）​​：记录当前执行位置（如 0 初始状态，1 暂停状态）。
# ​​当前值字段（_current）​​：保存 yield return 返回的值。
# ​​MoveNext() 方法​​：根据 _state 跳转到对应代码块执行，更新状态和当前值

# yield 的核心价值在于​  ​解耦数据生成  与  消费  过程​​，通过 状态机 机制  实现惰性求值  与  协作式多任务。
# 其适用于高并发I/O、大型数据处理、游戏逻辑等场景，显著提升代码可读性与资源利用率。
# 但需注意调试复杂性和线程安全问题，在性能敏感场景中需权衡状态机开销


# 特性	                  yield	              yield from
# 引入版本	           Python 2.2             	Python 3.3
# 功能	          生成器函数的基本构建块	  委托给另一个生成器或可迭代对象
# 代码复杂度	  需要手动迭代嵌套结构	         自动处理嵌套结构
# 异常处理	    需要手动处理子生成器的异常	       自动传递异常
# 返回值	           只能 yield 值	        可以获取子生成器的返回值

# 用于简化生成器的嵌套使用
def generator1():
    yield from [1, 2, 3]
    yield from (x for x in range(4, 7))

def generator1_equivalent():  #  等价操作
    for item in [1, 2, 3]:
        yield item
    for item in (x for x in range(4, 7)):
        yield item    

gen = generator1()
print(next(gen))  #  1
print(next(gen))  #  2
print(list(gen))  # 输出: [3, 4, 5, 6]

# 委托生成器：将生成操作委托给另一个生成器
def sub_generator():
    yield 1
    yield 2

def main_generator():
    yield from sub_generator()
    yield 3


mm = main_generator()
print(next(mm))  
print(next(mm))
print(next(mm))
# ------->  1 2 3
#  如果 next(main_generator()) 永远返回 1


# 协程：在异步编程中用于委托子协程
async def sub_coroutine():
    return 42

async def main_coroutine():
    result = await sub_coroutine()
    return result

# 简化代码：避免多层嵌套的 for 循环


# -----------------------------------
def sub_gen():
    yield 1
    yield 2
    print("*-------------")
    return "Done"

def delegator():
    result = yield from sub_gen()
    yield f"Sub-generator returned: {result}"

# print(list(delegator()))
# 输出: [1, 2, 'Sub-generator returned: Done']

gggx = delegator()

print(next(gggx))   #  1
print(next(gggx))   #  2
                    #  *-------------
print(next(gggx))   # Sub-generator returned: Done


# -------------------------------yield 实现数据处理管道----------------------------------------
def read_files(file_pattern):
    """读取多个文件"""
    for file_path in glob.glob(file_pattern):
        with open(file_path, 'r') as f:
            yield f.read()

def parse_lines(text):
    """解析文本行"""
    for line in text.split('\n'):
        yield line.strip()

def filter_comments(lines):
    """过滤注释行"""
    for line in lines:
        if not line.startswith('#'):
            yield line

# 构建处理管道
files = read_files('data/*.txt')
lines = parse_lines(''.join(files))
valid_lines = filter_comments(lines)

# 使用管道
for line in valid_lines:
    process_data(line)


# ----------------------------------yield----await------------------------------
# yield 用于生成器函数中，使函数变成一个生成器（可暂停恢复的迭代器）。 --  手动调度  --
        # --  next 拿数据   send  传入数据  --------------- 多用于处理数据
        #  send 首次调用必须是 send(None)  向生成器发送一个值，并返回yield的返回结果

# 特点
    # 用于同步代码
    # 创建生成器对象
    # 实现惰性求值
    # 用于协程（在 Python 早期版本中）

def coroutine():
    print("Start")
    x = yield 1  # 首次执行暂停于此，返回1
    print(f"Received: {x}")
    y = yield 2
c = coroutine()
print(c.send(None))  # 输出：Start → 1
print(c.send(10))    # 输出：Received: 10 → 2




# await 用于异步函数（async def）中，等待一个可等待对象（通常是协程）完成   ---  事件循环调度
        # 挂起当前协程  等待异步操作（如I/O）完成，期间事件循环可调度其他任务
        # -- 只能获取执行结果  ---- 必须依赖事件循环（如asyncio.run()）
        # --------------------------多用于高并发
# 特点
# 用于异步代码
# 只能在 async def 函数中使用
# 暂停当前协程的执行，直到 await 的对象完成   《=============
# 是异步编程的核心


# 协程（Coroutine）、Task、Future


# ----------------------------------awaitable 对象--------可等待对象--------------------------
# awaitable 对象：可以被 await 
# 可以被 await   的对象包括：
    # 实现了 __await__() 方法的对象
    # asyncio.Future 等
    # __await__ 必须返回一个 迭代器（通常是生成器），该迭代器最终生成一个值（即 await 的结果）。

# 当你在 await 表达式中使用某个对象时，Python 会调用它的 __await__ 方法来获取一个异步迭代器  result = await my_object
# 协程函数（async def）自动返回实现了 __await__ 的协程对象。


# 实际应用场景
    # 自定义异步操作：封装底层异步 I/O（如数据库查询、网络请求）。
    # 实现 Future 对象：类似 asyncio.Future，用于低级事件循环交互。
    # 高级异步模式：构建复杂异步工作流（如状态机、自定义调度器）。

# __await__ 和 __aiter__ 有何区别？
#     __await__：定义单次异步等待行为（await obj）。
#     __aiter__：定义异步迭代行为（async for x in obj）


# 为什么需要 yield from？
# yield from 将控制权交还事件循环，暂停当前协程，直到子操作完成（如 asyncio.sleep）


# ----------------------------------协程----------------------------------

# 核心原理
# 协程的核心原理是通过​​用户态的上下文切换​​实现多个执行流的协作式调度，避免操作系统线程切换的开销。
# 其本质是​​手动保存和恢复执行状态​​（寄存器、栈指针、程序计数器等），使单个线程内能交替执行多个任务

# 核心机制：上下文切换 保存状态  加载状态 （通常使用 汇编 切换寄存器）

# 栈管理
# 独占栈（Non-shared Stack）​​：每个协程预分配固定栈（通常KB级），简单但内存浪费。
# ​​共享栈（Shared Stack）​​：多个协程复用同一块栈内存，切换时需将栈内容拷贝到独立缓冲区，节省内存但增加拷贝开销

# 调度与协作
# 协程主动让出CPU（通过yield），而非被强制抢占。需在I/O阻塞或耗时操作前调用yield，切换至调度器。
# ​​事件驱动调度​​：调度器基于epoll监听I/O事件。当协程因I/O挂起时，调度器将其关联的fd加入epoll队列，事件就绪后恢复对应


# 核心价值：同步编码，异步性能

# 协程通过​​封装异步逻辑​​解决两类问题：
# ​​高并发I/O​​：网络服务器中，用协程处理海量连接（如Web服务），单线程内万级协程并发，避免线程同步锁。
# ​​简化异步代码​​：以同步风格写代码（如顺序调用recv、send），实际运行时通过Hook机制自动切换，底层转为异步I/O操作


# 限制与挑战
# ​​单线程瓶颈​​：协程本身不利用多核，需结合多进程（如Nginx模型）。
# ​​主动让出要求​​：协程需避免长时间占用CPU，否则阻塞调度。
# ​​调试复杂​​：状态切换导致堆栈信息断裂，增加调试难度


# C语言协程的核心是通过​​用户态上下文切换​​（寄存器/栈保存与恢复）实现​​协作式多任务​​，
# 结合​​事件驱动调度​​将I/O阻塞点转化为协程切换点，最终以同步代码风格获得异步性能。
# 其轻量级（KB级内存）、高并发（单线程万级协程）特性，使其成为I/O密集型应用（网络服务、游戏服务器）的高效解决方案



# python协程进化历程
# 生成器（yield）​​： 早期通过yield暂停/恢复执行，模拟协程行为（例：生产者-消费者模型）。
# ​​Greenlet​​：       提供显式切换（switch()），但需手动管理切换时机。
# ​​Gevent​​：         基于Greenlet封装，通过monkey.patch_all()替换标准库的阻塞I/O（如socket），实现​​自动切换​​（遇I/O阻塞时跳转）。
# ​​asyncio与async/await​​（Python 3.5+）：
                    # async def定义协程，await暂停等待异步操作（如I/O）。
                    # 事件循环（Event Loop）调度任务，实现高效并发。



import asyncio

# 定义一个简单的协程函数
async def simple_coroutine(name, delay):
    print(f"{name} 开始执行，将等待 {delay} 秒")
    await asyncio.sleep(delay)  # 模拟IO操作
    print(f"{name} 完成执行")
    return f"{name} 的结果"

# 主协程函数
async def main():
    # 创建多个协程任务
    task1 = asyncio.create_task(simple_coroutine("协程A", 2))
    task2 = asyncio.create_task(simple_coroutine("协程B", 1))
    task3 = asyncio.create_task(simple_coroutine("协程C", 3))
    
    # 等待所有任务完成并收集结果
    results = await asyncio.gather(task1, task2, task3)
    print("所有协程完成:", results)

# 运行事件循环
asyncio.run(main())


# ------------------------------------用到的api----------------------------------------------
# async def 定义协程函数
# await 用于挂起协程的执行，等待异步操作完成
# 核心方法：run(), create_task(), gather(), wait(), sleep()。
# 同步工具：Lock, Event, Queue。
# 线程交互：run_in_executor(), to_thread()。

#  ---------------------事件循环管理------------------- 
# asyncio.get_event_loop()：获取当前事件循环（旧版本常用，Python 3.10+ 建议用 asyncio.run）。
loop = asyncio.get_event_loop()
# loop.run_until_complete(future)：运行直到 future（协程）完成。
loop.run_until_complete(main())
# loop.run_forever()：无限运行事件循环（需手动停止）。
# loop.close()：关闭事件循环。




#  ---------------------协程与任务-------------------  
# asyncio.create_task(coro)  将协程包装为 Task 并调度执行
task = asyncio.create_task(coro())
# asyncio.gather(*coros, return_exceptions=False) 并发运行多个协程，返回结果列表
results = await asyncio.gather(coro1(), coro2())
# asyncio.run(coro, *, debug=False)运行一个协程并管理事件循环
asyncio.run(main())
# asyncio.wait(coros, *, timeout=None, return_when=ALL_COMPLETED) 等待多个协程完成，返回 (done, pending) 集合。
done, pending = await asyncio.wait([coro1(), coro2()])
# asyncio.shield(aw) 保护协程不被取消（类似“防弹衣”）。



#  ---------------------异步等待-------------------    
# asyncio.sleep(delay, result=None)  异步等待指定时间（非阻塞）。
await asyncio.sleep(1)
# asyncio.wait_for(aw, timeout)   带超时的等待。
try:
    await asyncio.wait_for(coro(), timeout=2)
except asyncio.TimeoutError:
    print("Timeout!")





#  ---------------------跨线程调度-------------------    
# asyncio.Lock()  异步互斥锁。
lock = asyncio.Lock()
async with lock:
    # 临界区

# asyncio.Event()   事件通知机制
event = asyncio.Event()
await event.wait()  # 等待事件触发
event.set()         # 触发事件

# asyncio.Queue()  异步队列（生产者-消费者模型）。
queue = asyncio.Queue()
await queue.put(item)
item = await queue.get()





#  ---------------------跨线程调度-------------------
# loop.call_soon(callback, *args)  在事件循环中立即调度回调。
# loop.run_in_executor(executor, func, *args)  在线程池中运行同步函数（避免阻塞事件循环）。
await loop.run_in_executor(None, time.sleep, 1)



#  ---------------------其他工具-------------------
# asyncio.to_thread(func, *args)  将同步函数放到独立线程中运行（Python 3.9+）
await asyncio.to_thread(blocking_func, arg)

# asyncio.timeout(delay)  上下文管理器，设置超时（Python 3.11+）。
async with asyncio.timeout(2):
    await coro()



# loop.call_soon(print, "立即执行")
# loop.call_later(2, print, "2秒后执行")

# -------------------------协程让出所有权----------------------------------
# 当执行到await
#     会挂起当前协程的执行
#     将控制权返回给事件循环
#     事件循环可以调度其他协程执行
#     当await后面的对象(通常是另一个协程)完成时，原协程恢复执行

# 显式调用asyncio.sleep()

# 任务切换点：即使没有显式await，事件循环也会在某些点切换任务：
#     当前协程执行足够长时间(防止一个协程独占CPU)
#     使用asyncio.shield()
#     使用asyncio.wait_for()设置超时

# 发生异常（没捕获的）



# 最佳实践
#     在协程中避免长时间运行的同步操作
#     所有I/O操作都应使用异步版本并配合await
#     需要CPU密集型操作时考虑使用loop.run_in_executor()



# await x实际上会被Python编译为： yield from x.__await__()
# 而x.__await__()通常会yield一个Future，所以Task需要通过send()来驱动这个流程





# -------------------过时写法-----------------------
@asyncio.coroutine
def step1():
    print("执行步骤1")
    yield from asyncio.sleep(1)
    return "步骤1结果"

@asyncio.coroutine
def workflow():
    result1 = yield from step1()
    return result1

loop = asyncio.get_event_loop()
result = loop.run_until_complete(workflow())
print("工作流结果:", result)
loop.close()

# -------------------最新写法-----------------------
async def modern_coroutine():
    print("现代协程")
    await asyncio.sleep(1)
    return "结果"

asyncio.run(modern_coroutine())



# contextlib.asynccontextmanager 是 Python 3.7+ 提供的装饰器，用于将异步生成器函数转换为异步上下文管理器。
import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_resource_manager():
    print("初始化资源")
    yield "资源句柄"  # 进入 `async with` 时返回
    print("清理资源")

async def use_resource():
    async with async_resource_manager() as resource:
        print(f"使用资源: {resource}")
        await asyncio.sleep(1)

asyncio.run(use_resource())

# 初始化资源
# 使用资源: 资源句柄
# 清理资源




# -------------------------数据库请求失败后重试-------------------------------------
import asyncio
from functools import wraps

def retry_on_failure(max_retries=3, delay=1):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(1, max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries:
                        raise
                    print(f"第 {attempt} 次失败，{delay} 秒后重试...")
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

@retry_on_failure(max_retries=3, delay=1)
async def unreliable_request():
    import random
    if random.random() < 0.7:  # 70% 概率失败
        raise ValueError("请求失败！")
    return "请求成功！"

asyncio.run(unreliable_request())


# -------------------------超时控制-------------------------------------
import asyncio
from functools import wraps

def timeout(seconds):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await asyncio.wait_for(func(*args, **kwargs), timeout=seconds)
            except asyncio.TimeoutError:
                print(f"函数 {func.__name__} 超时（{seconds} 秒）")
                raise
        return wrapper
    return decorator

@timeout(2)
async def long_running_task():
    await asyncio.sleep(3)  # 故意超时
    return "任务完成"

try:
    asyncio.run(long_running_task())
except asyncio.TimeoutError:
    print("捕获超时异常")



# --------------------------------------------------------------



import selectors
import time
from collections import deque
import socket
import sys

# Future 代表一个异步操作的最终结果，是协程和事件循环之间的桥梁。
class Future:
    def __init__(self):
        # 存储异步操作的结果（_result）
        # 管理完成回调（_callbacks）
        # 跟踪完成状态（_done）
        self._result = None
        self._callbacks = []
        self._done = False

    def result(self):
        if not self._done:
            raise RuntimeError("Future not done yet")
        return self._result

    def add_done_callback(self, fn):
        if self._done:
            fn(self)
        else:
            self._callbacks.append(fn)

    def set_result(self, result):
        if self._done:
            raise RuntimeError("Future already done")
        self._result = result
        self._done = True
        for callback in self._callbacks:
            callback(self)

    def __await__(self):
        # await x实际上会被Python编译为： yield from x.__await__()
        # 而x.__await__()通常会yield一个Future，所以Task需要通过send()来驱动这个流程
        yield self
        return self.result()

# Task 负责驱动协程的执行，是协程的包装器。
class Task(Future):  # 让Task继承自Future
    def __init__(self, coro, loop):
        super().__init__()
        self.coro = coro
        self.loop = loop
        self._step()

    def _step(self, future=None):
        try:
            # 在 Python 中，使用 async 定义的协程函数（coroutine function）
            # 在被调用后返回的协程对象（coroutine object）​​本身并没有 send 方法​​。
            # 不过，协程对象的底层实现是通过生成器（generator）机制实现的，
            # 而生成器是有 send 方法的。
            if future is None:
                result = self.coro.send(None)  # 首次启动协程 
            else:
                result = self.coro.send(future.result())
        except StopIteration as exc:
            self.set_result(exc.value if exc.value is not None else None)  # 恢复协程并传入值
            return

        if isinstance(result, Future):
            result.add_done_callback(self._step)

# 事件循环是异步编程的核心，负责调度和执行所有任务。
class EventLoop:
    def __init__(self):
        # Windows socket 初始化
        if sys.platform == 'win32':
            self._wsa_initialized = hasattr(socket, 'WSAStartup')
            if self._wsa_initialized:
                wsa_version = 0x202  # Version 2.2
                socket.WSAStartup(wsa_version)

        self._selector = selectors.DefaultSelector()
        self._ready = deque()
        self._stopping = False
        self._scheduled = []
        self._current_handles = set()

    def call_soon(self, callback, *args):
        self._ready.append((callback, args))

    def call_later(self, delay, callback, *args):
        t = time.time() + delay
        self._scheduled.append((t, callback, args))
        self._scheduled.sort(key=lambda x: x[0])

    def create_task(self, coro):
        task = Task(coro, self)
        return task

    def run_forever(self):
        while not self._stopping:
            if not self._ready and not self._scheduled:
                break
            self._run_once()

    def _run_once(self):
        # 处理准备就绪的回调
        ntodo = len(self._ready)
        for _ in range(ntodo):
            callback, args = self._ready.popleft()
            callback(*args)

        # 处理到期的定时任务
        now = time.time()
        while self._scheduled and self._scheduled[0][0] <= now:
            t, callback, args = self._scheduled.pop(0)
            self.call_soon(callback, *args)

        # 计算超时时间
        timeout = None
        if self._scheduled:
            timeout = max(0, self._scheduled[0][0] - time.time())
        if self._ready:
            timeout = 0

        # 处理IO事件
        if not self._ready:
            try:
                events = self._selector.select(timeout)
                for key, mask in events:
                    callback = key.data
                    if key.fileobj in self._current_handles:
                        self.call_soon(callback, key.fileobj)
            except OSError as e:
                if getattr(e, 'winerror', None) != 10022:  # 忽略Windows上的无效参数错误
                    raise

    def stop(self):
        self._stopping = True

    def add_reader(self, fd, callback):
        self._selector.register(fd, selectors.EVENT_READ, callback)
        self._current_handles.add(fd)

    def remove_reader(self, fd):
        try:
            self._selector.unregister(fd)
            self._current_handles.discard(fd)
        except KeyError:
            pass

    def run_until_complete(self, coro):
        task = self.create_task(coro)
        self.run_forever()
        return task.result()

    def close(self):
        self._selector.close()
        if hasattr(self, '_wsa_initialized') and self._wsa_initialized:
            socket.WSACleanup()


def sleep(delay):
    future = Future()
    loop.call_later(delay, lambda: future.set_result(None))
    return future


# 示例使用
async def example_coroutine():
    print("开始协程")
    await sleep(1)
    print("1秒后")
    await sleep(2)
    print("又2秒后")
    return "完成"


# 创建并运行事件循环
loop = EventLoop()
try:
    result = loop.run_until_complete(example_coroutine())
    print("协程结果:", result)
finally:
    loop.close()






